package com.bw.gmall.realtime.app.dwd;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import com.bw.gmall.realtime.utils.MysqlUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderRefund {
    public static void main(String[] args) throws Exception {
        //todo 1.初始化上下文  设置并行度
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        // todo 获取配置对象
        Configuration configuration = tenv.getConfig().getConfiguration();
        // todo 为表关联时状态中存储的数据设置过期时间
        configuration.setString("table.exec.state.ttl","5s");


        tenv.executeSql("create table topic_db ( " +
                "`database` string,  " +
                "`table` string,  " +
                "`type` string,  " +
                "`data` map<string,string>,  " +
                "`old` map<string,string>,  " +
                "`proc_time` as proctime(),  " +
                "`ts` string  " +
                ")" + MyKafkaUtil.getKafkaDDL("topic_db","order_refund"));

        // TODO 4. 读取退单表数据
        Table orderRefundInfo = tenv.sqlQuery("select " +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['order_id'] order_id, " +
                "data['sku_id'] sku_id, " +
                "data['refund_type'] refund_type, " +
                "data['refund_num'] refund_num, " +
                "data['refund_amount'] refund_amount, " +
                "data['refund_reason_type'] refund_reason_type, " +
                "data['refund_reason_txt'] refund_reason_txt, " +
                "data['create_time'] create_time, " +
                "proc_time, " +
                "ts   " +
                "from topic_db " +
                "where `table` = 'order_refund_info' " +
                "and  `type` = 'insert' ");

        tenv.createTemporaryView("order_refund_info",orderRefundInfo);
//        tenv.sqlQuery("select * from order_refund_info").execute().print();


        //todo 5.读取订单表数据,筛选退单数据
        Table orderInfoTable = tenv.sqlQuery("select " +
                "data['id'] id," +
                "data['province_id'] province_id," +
                "`old`  " +
                "from topic_db " +
                "where `table` =  'order_info' " +
                "and `type` = 'update' " +
                "and data['order_status']='1005' " +
                "and `old`['order_status'] is not null ");

        tenv.createTemporaryView("order_info_refund",orderInfoTable);

        //todo 6.建立字典表
        tenv.executeSql(MysqlUtil.getBaseDicLookUpDDL());

        //todo 7.三表关联

        Table resTable = tenv.sqlQuery("select " +
                "ri.id, " +
                "ri.user_id, " +
                "ri.order_id, " +
                "ri.sku_id, " +
                "ri.refund_type, " +
                "ri.refund_num, " +
                "ri.refund_amount, " +
                "ri.refund_reason_type, " +
                "ri.refund_reason_txt, " +
                "ri.create_time, " +
                "oi.province_id, " +
                "type_dic.dic_name, " +
                "reason_dic.dic_name, " +
                "date_format(ri.create_time,'yyyy-MM-dd') date_id," +
                "ri.ts," +
                "current_row_timestamp() row_op_ts " +
                "from order_refund_info ri " +
                "join  " +
                "order_info_refund oi " +
                "on ri.order_id = oi.id " +
                "join " +
                "base_dic for system_time as of ri.proc_time as type_dic " +
                "on ri.refund_type = type_dic.dic_code " +
                "join " +
                "base_dic for system_time as of ri.proc_time as reason_dic " +
                "on ri.refund_reason_type = reason_dic.dic_code");
        tenv.createTemporaryView("res_table",resTable);

        //todo 8.建立表
        tenv.executeSql("create table dwd_trade_order_refund (" +
                "id string, " +
                "user_id string, " +
                "order_id string, " +
                "sku_id string, " +
                "refund_type_code string, " +
                "refund_num string, " +
                "refund_amount string, " +
                "refund_reason_type_code string, " +
                "refund_reason_txt string, " +
                "create_time string, " +
                "province_id string, " +
                "refund_type_name string, " +
                "refund_reason_type_name string, " +
                "date_id string, " +
                "ts string, " +
                "row_op_ts timestamp_ltz(3) " +
                ")"+ MyKafkaUtil.getKafkaSinkDDL("dwd_trade_order_refund"));

        tenv.executeSql("insert into dwd_trade_order_refund select * from res_table");
        tenv.sqlQuery("select * from dwd_trade_order_refund").execute().print();




//        env.execute();
    }
}
